package basicutil.common;

import com.sun.javafx.tk.Toolkit;
import javafx.application.Platform;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

@Slf4j
public class ThreadUtil {
    public static void runLater(Runnable runnable) {
        pool.execute(runnable);
    }

    public static void runLaterSafely(Runnable runnable) {
        pool.execute(new SafeRunnable(runnable));
    }

    public static void waitForPlatform(Runnable runnable) {
        if (Toolkit.getToolkit().isFxUserThread()) {
            runnable.run();
            return;
        }
        CountDownLatch countDownLatch = new CountDownLatch(1);
        Platform.runLater(
                () -> {
                    try {
                        runnable.run();
                    } finally {
                        countDownLatch.countDown();
                    }
                }
        );
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static <E> void drainQueue(BlockingQueue<E> queue, Consumer<E> consumer) {
        pool.execute(new SafeRunnable(() -> {
            while (true) {
                try {
                    E e = queue.take();
                    log.debug("队列{}，获取到消息：{},consumer:{}", queue.hashCode(), e, consumer);
                    consumer.accept(e);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }));
    }

    @SneakyThrows
    public static void sleep(long millionSeconds) {
        Thread.sleep(millionSeconds);
    }

    public static void runRepeat(BlockingQueue<String> q, Runnable r) {
        AtomicBoolean running = new AtomicBoolean(false);

        drainQueue(q, ss -> {
            synchronized (running) {
                if (running.get()) {
                    running.set(false);

                } else {
                    running.set(true);
                    running.notifyAll();
                }
            }
        });
        runLaterSafely(() -> {
            while (true) {
                while (!running.get()) {
                    try {
                        System.out.println("wait on running.get()");
                        synchronized (running) {
                            running.wait();
                        }

                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
// 放入循环体
                r.run();
            }
        });
    }

    private static class SafeRunnable implements Runnable {
        private final Runnable r;

        SafeRunnable(Runnable r) {
            this.r = r;
        }

        @Override
        public void run() {
            try {
                r.run();
            } catch (Throwable e) {
                log.error("------exception raised by Runnable :{}------", r, e);
            }
        }
    }

    private static final ExecutorService pool = Executors.newCachedThreadPool(new ThreadFactory() {
        final AtomicInteger threadCounter = new AtomicInteger();

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            log.warn("create thread: {}", thread);
            thread.setName("|my thread-" + threadCounter.incrementAndGet());
            thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                @Override
                public void uncaughtException(Thread t, Throwable e) {
                    log.error("------uncaughtException thread :{}------", t, e);
                }
            });
            return thread;
        }
    });


}
